package cn.doitedu.cn.doitedu.rtdw.data_sync;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job01_UserTable2Hbase {
    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 1.建表，映射mysql中的user表
        tenv.executeSql(
                " create table ums_member_mysql(     "+
                        "    id bigint,                       "+
                        "    username string,                 "+
                        "    phone string,                    "+
                        "    status int,                      "+
                        "    create_time timestamp(3),        "+
                        "    gender int,                      "+
                        "    birthday date,                   "+
                        "    city string,                     "+
                        "    job string,                      "+
                        "    source_type int,                 "+
                        "    PRIMARY KEY (id) NOT ENFORCED    "+
                        " ) with (                            "+
                        " 'connector' = 'mysql-cdc',          "+
                        " 'hostname' = 'doitedu'   ,          "+
                        " 'port' = '3306'          ,          "+
                        " 'username' = 'root'      ,          "+
                        " 'password' = 'root'      ,          "+
                        " 'database-name' = 'realtimedw',     "+
                        " 'table-name' = 'ums_member'         "+
                        " )                                   "

        );

       //tenv.executeSql("select * from ums_member_mysql ").print();



        // 2.建表，映射hbase中的目标表
        tenv.executeSql(
                " create table ums_member_hbase(           "+
                        "   username string,                       "+
                        " 	f row<id bigint,                       "+
                        " 	phone string,                          "+
                        " 	status int,                            "+
                        " 	create_time timestamp(3),              "+
                        " 	gender int,                         "+
                        " 	birthday date,                         "+
                        "     city string,                         "+
                        "     job string,                          "+
                        "     source_type int>,                    "+
                        " 	PRIMARY KEY (username) NOT ENFORCED    "+
                        " ) with (                                 "+
                        "  'connector' = 'hbase-2.2',              "+
                        "  'table-name' = 'dim_ums_member',        "+
                        "  'zookeeper.quorum' = 'doitedu:2181'     "+
                        " )                                        "
        );


        // 3.写sql
        tenv.executeSql("insert into ums_member_hbase select username, row(id,phone,status,create_time,gender,birthday,city,job,source_type) as f from  ums_member_mysql");


    }
}
